package com.example.kafka.service;

import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Service;

import java.util.concurrent.CompletableFuture;

/**
 * Kafka生产者服务
 */
@Service
@Slf4j
public class KafkaProducerService {

    private final KafkaTemplate<String, String> kafkaTemplate;

    public KafkaProducerService(KafkaTemplate<String, String> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }

    /**
     * 发送消息到指定主题
     *
     * @param topic   目标主题
     * @param message 消息内容
     */
    public void sendMessage(String topic, String message) {
        log.info("Producing message to topic {}: {}", topic, message);
        // 发送消息，不关心结果
        kafkaTemplate.send(topic, message);
    }

    /**
     * 发送消息并异步处理结果
     *
     * @param topic   目标主题
     * @param key     消息键
     * @param message 消息内容
     */
    public void sendMessageWithCallback(String topic, String key, String message) {
        log.info("Producing message to topic {} with key {}: {}", topic, key, message);

        kafkaTemplate.executeInTransaction(kt -> {
            // 发送消息并添加回调
            CompletableFuture<SendResult<String, String>> future = kt.send(topic, key, message);

            future.thenAccept(result -> {
                log.info("Sent message successfully to topic {} with offset {}",
                        result.getRecordMetadata().topic(),
                        result.getRecordMetadata().offset());
            }).exceptionally(ex -> {
                log.error("Failed to send message to topic {}", topic, ex);
                return null;
            });
            return true;
        });

    }
}